package com.shadow.disnet.netty.server;

import com.shadow.disnet.disrutor.DisruptorExecutor;
import com.shadow.disnet.netty.ILog;
import com.shadow.disnet.netty.Life;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

public class NettyServer implements ILog, Life {
    private final Logger logger = LoggerFactory.getLogger(NettyServer.class);
    private int PORT = 10010;
    private ServerBootstrap config;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private ChannelGroup group;
    private final LinkedBlockingQueue<Channel> channels;
    private ExecutorService normal;
    private ScheduledExecutorService schdule;

    private DisruptorExecutor[] disruptor;

    /**
     * backlog
     */
    public NettyServer() {
        group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        channels = new LinkedBlockingQueue<>();
        ThreadFactory bf = new DefaultThreadFactory("boss线程", false, 5);
        bossGroup = new NioEventLoopGroup(1, bf);
        ThreadFactory wf = new DefaultThreadFactory("worker线程", false, 5);
        workerGroup = new NioEventLoopGroup(16, wf);
        config = new ServerBootstrap();
        config.group(bossGroup, workerGroup);
        config.option(ChannelOption.SO_BACKLOG, 128); //客户端连接队列
        config.option(ChannelOption.TCP_NODELAY, true);
        config.option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT);//自适应
        config.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT); //缓冲池
        config.handler(new LoggingHandler(LogLevel.DEBUG));
        config.channel(NioServerSocketChannel.class).handler(new ChannelInitializer<NioServerSocketChannel>() {
            @Override
            protected void initChannel(NioServerSocketChannel ch) throws Exception {
                info("初始化ServerSocketChannel=%s", ch);
            }

        }).childHandler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                info("初始化SocketChannel=%s", ch);
                ch.pipeline().addLast("encode", new StringEncoder());
                ch.pipeline().addLast("decode", new StringDecoder());
                ch.pipeline().addLast("InHandler", new MySeverInBoundHandler(group, channels, NettyServer.this));
                ch.pipeline().addLast("OutHandler", new MyServerOutBoundHandler());
            }
        });
        ThreadFactory uf = new DefaultThreadFactory("用户线程", false, 5);
        ThreadFactory uff = new DefaultThreadFactory("用户定时线程", false, 5);
        normal = new ThreadPoolExecutor(
                4, 16,
                60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(16), uf);
        schdule = new ScheduledThreadPoolExecutor(16, uff);
        disruptor = new DisruptorExecutor[16];
        for (int i = 0; i < 16; i++) {
            disruptor[i] = new DisruptorExecutor(1024, "用户快速线程");
        }

    }

    public DisruptorExecutor getExecutor(long id) {
        return disruptor[Math.abs((int) id % disruptor.length)];
    }

    public void start() {
//        schdule.scheduleAtFixedRate(()->{
//            info("在线人数：%s---%s", group.size(),channels.size());
//        },0,10,TimeUnit.SECONDS);

//        schdule.scheduleAtFixedRate(()->{
//                Channel remove;
//                try {
//                    remove = channels.take(); //如果没有就阻塞
//                    Channel finalRemove = remove;
//                    group.close(channel -> finalRemove == channel).addListener(f -> {
//                        if (f.isSuccess()) {
//                            info("踢下线一个：%s", finalRemove);
//                            info("踢人服务：%s---%s", group.size(),channels.size());
//                        }
//                    });
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//
//        },10,1,TimeUnit.SECONDS);
        ChannelFuture bind = config.bind(PORT).addListener(future -> {
            if (future.isSuccess()) {
                info("服务器绑定端口%s成功", PORT);
            }
        });
        bind.channel().closeFuture().addListener(cf -> {
            if (cf.isSuccess()) {
                close();
                info("通道关闭，优雅的关闭服务器线程池");
            }
        });

    }

    public void close() {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

    @Override
    public Logger getLogger() {
        return logger;
    }

    public static void main(String[] args) {
        NettyServer nettyServer = new NettyServer();
        nettyServer.start();
    }

}
